package com.zongxingh.falcon.mq.consumer;

import com.zongxingh.falcon.mq.config.MqProperties;
import com.zongxingh.falcon.mq.constant.MqType;
import com.zongxingh.falcon.mq.consumer.impl.RabbitMqConsumer;
import com.zongxingh.falcon.mq.consumer.impl.RocketMqConsumer;
import com.zongxingh.falcon.mq.utils.MqUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;

/**
 * Introduction:  <简介>
 * Package Name:  com.zongxingh.falcon.mq.consumer
 * Project Name:  mq
 * Author:  zongxingh
 * Email： zongxingh@163.com
 * Tel: 157*****778
 * Create Time:  2020/9/22 15:03
 */
public class MQConsumerFactory {
    private static final Logger LOGGER = LoggerFactory.getLogger(MQConsumerFactory.class);

    private static Map<String, MqConsumer> comsumerMap = new HashMap<>();

    public static MqConsumer createMqConsumer(String consumerGroup, String topic, MqProperties properties, MsgConsume msgConsume) {
        if (null == properties.getMqType() || "".equals(properties.getMqType())){
            LOGGER.error("获得消息消费者，当前没有配置消息队列类型。");
            return null;
        }
        else if (MqType.ROCKET_MQ.getCode().equals(properties.getMqType())) {
            return getRocketMqConsumer(consumerGroup, topic, properties, msgConsume);
        }
        else if (MqType.RABBIT_MQ.getCode().equals(properties.getMqType())) {
            return getRabbitMqConsumer(consumerGroup, topic, properties, msgConsume);
        }
        else {
            LOGGER.warn("获得消息消费者，当前不支持该消息队列，消息队列类型为：{}", properties.getMqType());
            return null;
        }
    }

    private static MqConsumer getMqComsumer(String consumerName) {
        if (comsumerMap.size() == 0) {
            return null;
        }
        return comsumerMap.get(consumerName);
    }

    /**
     * 获得 RocketMq Consumer
     *
     * @param consumerGroup
     * @param topic
     * @param msgConsume
     * @return
     */
    private static MqConsumer getRocketMqConsumer(String consumerGroup, String topic, MqProperties properties, MsgConsume msgConsume) {
        String consumerName = MqUtils.genMqConsumerName(consumerGroup,topic);
        MqConsumer mqConsumer = getMqComsumer(consumerName);
        if (null == mqConsumer) {
            RocketMqConsumer rocketMqConsumer = new RocketMqConsumer(consumerGroup, topic, properties, msgConsume);
            comsumerMap.put(rocketMqConsumer.getName(), rocketMqConsumer);
            return rocketMqConsumer;
        }
        return mqConsumer;
    }

    /**
     * 获得RabbitMq Consumer
     *
     * @param consumerGroup
     * @param topic
     * @param msgConsume
     * @return
     */
    private static MqConsumer getRabbitMqConsumer(String consumerGroup, String topic, MqProperties properties, MsgConsume msgConsume) {
        MqConsumer mqConsumer = getMqComsumer(consumerGroup);
        if (null == mqConsumer) {
            RabbitMqConsumer rabbitMqConsumer = new RabbitMqConsumer();
            comsumerMap.put(consumerGroup, rabbitMqConsumer);
            return rabbitMqConsumer;
        }
        return mqConsumer;
    }

}
